package com.gitee.xhs.saveAndModifyingPointWriter.windowState;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.dao.ModifyingPointDao;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.util.SavePointApiUtil;
import com.gitee.xhs.windowState.TransactionOneMinuteCounter;
import com.gitee.xhs.windowState.WindowStateConstant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.state.api.OperatorTransformation;
import org.apache.flink.state.api.SavepointWriter;
import org.apache.flink.state.api.StateBootstrapTransformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Timestamp;
import java.time.Duration;

import static com.gitee.xhs.StateConstant.CHECK_POINT_BASE_PATH;
import static com.gitee.xhs.saveAndModifyingPointWriter.SavePointWriterConstant.MAX_PARALLELISM;
import static com.gitee.xhs.saveAndModifyingPointWriter.SavePointWriterConstant.MODIFYING_UID_PREFIX;
import static com.gitee.xhs.windowState.WindowStateConstant.CUSTOM_WINDOW_STATE_DIR;
import static com.gitee.xhs.windowState.WindowStateConstant.WINDOW_STATE_UID;

/**
 * @author BigHead
 * @date 2022/7/12 14:29
 * @description WindowStateNewSavePointsMain
 */
public class WindowStateNewSavePointsAndModifyingMain {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.simpleEnv(new EnvironmentConfig(null));

        DataStream<Transaction> transactionDataStream = streamExecutionEnvironment.fromElements(
                        new Transaction(1, Timestamp.valueOf("1970-01-01 00:00:01").getTime(), 0.1),
                        new Transaction(2, Timestamp.valueOf("1970-01-01 00:00:01").getTime(), 0.1),
                        new Transaction(3, Timestamp.valueOf("1970-01-01 00:00:01").getTime(), 0.1),
                        new Transaction(4, Timestamp.valueOf("1970-01-01 00:00:01").getTime(), 0.1),
                        new Transaction(5, Timestamp.valueOf("1970-01-01 00:00:01").getTime(), 0.1),

                        new Transaction(1, Timestamp.valueOf("1970-01-01 00:00:50").getTime(), 0.1),
                        new Transaction(2, Timestamp.valueOf("1970-01-01 00:00:50").getTime(), 0.1),
                        new Transaction(3, Timestamp.valueOf("1970-01-01 00:00:50").getTime(), 0.1),
                        new Transaction(4, Timestamp.valueOf("1970-01-01 00:00:50").getTime(), 0.1),
                        new Transaction(5, Timestamp.valueOf("1970-01-01 00:00:50").getTime(), 0.1)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(
                                        (transaction, timestamp) -> transaction.getTimestamp()
                                )
                );

        String newSavePointDirPathName = SavePointApiUtil.getNewSavePointWriterJobPath(CHECK_POINT_BASE_PATH + CUSTOM_WINDOW_STATE_DIR);

        StateBootstrapTransformation<Transaction> transformation = OperatorTransformation
                .bootstrapWith(transactionDataStream)
                .keyBy(Transaction::getAccountId)
                .window(WindowStateConstant.TUMBLING_EVENT_TIME_WINDOWS_ONE_MINUTE)
                .aggregate(new TransactionOneMinuteCounter());

        SavepointWriter
                .newSavepoint(new HashMapStateBackend(), MAX_PARALLELISM)
                .withOperator(WINDOW_STATE_UID, transformation)
                .write(newSavePointDirPathName);

        ModifyingPointDao modifyingPointDao = SavePointApiUtil.getModifyingPath(CHECK_POINT_BASE_PATH + CUSTOM_WINDOW_STATE_DIR);

        SavepointWriter
                .fromExistingSavepoint(modifyingPointDao.getOldPath(), new HashMapStateBackend())
                .withOperator(WINDOW_STATE_UID + MODIFYING_UID_PREFIX, transformation)
                .write(modifyingPointDao.getNewPath());

        streamExecutionEnvironment.execute("WindowStateNewSavePointsMain");
    }

}
